package cn.zkrt.iotms.module.data.service.impl;


import cn.zkrt.iotms.config.InfluxDBConfig;
import cn.zkrt.iotms.module.data.dto.AttributeData;
import cn.zkrt.iotms.module.data.dto.InfluxDto;
import cn.zkrt.iotms.module.data.dto.InfluxNewDto;
import cn.zkrt.iotms.module.data.service.InfluxdbService;
import cn.zkrt.iotms.utils.DateUtils;
import cn.zkrt.iotms.utils.StringUtils;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.QueryApi;
import com.influxdb.query.FluxTable;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

@Service("influxdbService")
@Slf4j
public class InfluxdbServiceImpl implements InfluxdbService {

    @Autowired
    private InfluxDBClient influxDBClient;

    @Autowired
    private InfluxDBConfig influxDBConfig;


    @Override
    public Object query(String flux, String code, Map<String, Object> map) {
        /*if (code.equals("CJ_ELECTRIC_CONSUME")) {
            return cjElectricConsume(flux, map);
        } else if (code.equals("CJ_SENSOR_HISTORY")) {
            return cjSensorHistory(flux, map);
        }*/
        return null;
    }


    @Override
    public Object query(String flux, String code) {
        return null;
    }

    @Override
    public Map<String, List<AttributeData>> getHistories(InfluxDto influxDto, Date start, Date end) {

        QueryApi queryApi = influxDBClient.getQueryApi();

        List<String> parametrizedQuery = new ArrayList<>();

        parametrizedQuery.add(String.format("from(bucket: \"%s\")", influxDto.getBucket()));
        parametrizedQuery.add(String.format("range(start: %s, stop: %s)", DateUtils.subtract8Hours(start) ,DateUtils.subtract8Hours(end))); // toInfluxDate 就是给小时减了8小时，其实在前端处理更合理，只是我们系统没有国际化的需求，算是给前端偷懒了
        if (StringUtils.isNotEmpty(influxDto.getDevSn())){
            parametrizedQuery.add(String.format("filter(fn: (r) => r[\"_measurement\"] == \"%s\")", influxDto.getDevSn()));
        }


        if (StringUtils.isNotEmpty(influxDto.getDevSn())){
            /*String devSnFilter = influxDto.getDevSn().stream().map(devSn -> String.format("r[\"devSn\"] == \"%s\"", devSn)).collect(Collectors.joining(" or "));

            System.out.println("拿到了:"+devSnFilter);*/
            parametrizedQuery.add(String.format("filter(fn: (r) => r[\"devSn\"] == \"%s\")", influxDto.getDevSn()));
            //parametrizedQuery.add(String.format("devSn(fn: (r) => %s)", devSnFilter));
        }

        if (influxDto.getFields()!=null&&influxDto.getFields().size()>0){
            String filedFilter = influxDto.getFields().stream().map(field -> String.format("r[\"_field\"] == \"%s\"", field)).collect(Collectors.joining(" or "));
            parametrizedQuery.add(String.format("filter(fn: (r) => %s)", filedFilter));
        }

        /**
         * 排序 按时间倒序
         */
        parametrizedQuery.add("sort(columns: [\"_time\"], desc: true)");

        /**
         * 分页
         */
        //parametrizedQuery.add(String.format("limit(n: %s, offset: %s)", pageSize, pageSize * (pageNum - 1)));

        Map<String, List<AttributeData>> result = new HashMap<>();

        List<FluxTable> tables = queryApi.query(String.join(" |> ", parametrizedQuery));

        for (FluxTable table : tables) {

            List<List> list = table.getRecords().stream().map(record ->
                    Arrays.asList(record.getValue(), record.getTime().atZone(ZoneId.systemDefault()).toLocalDateTime())  // 日期直接转成本地的，前端拿去直接用
            ).collect(Collectors.toList());

            ArrayList<AttributeData> dataList = new ArrayList<>();
            list.forEach(s->{
                AttributeData data = new AttributeData();
                data.setValue(s.get(0).toString());
                data.setTheTime(s.get(1).toString());
                dataList.add(data);
            });


            result.put(table.getRecords().get(0).getField(), dataList);
        }

        return result;

    }



    @Override
    public Map<String, List<AttributeData>> getHistories(InfluxDto influxDto) {

        QueryApi queryApi = influxDBClient.getQueryApi();

        List<String> parametrizedQuery = new ArrayList<>();

        parametrizedQuery.add(String.format("from(bucket: \"%s\")", influxDto.getBucket()));
        // toInfluxDate 就是给小时减了8小时，其实在前端处理更合理，只是我们系统没有国际化的需求，算是给前端偷懒了
        parametrizedQuery.add(String.format("range(start: %s, stop: %s)",DateUtils.subtract8Hours(influxDto.getStartTime()) ,DateUtils.subtract8Hours(influxDto.getEndTime())));
        // _measurement 就是devSn
        if (StringUtils.isNotEmpty(influxDto.getDevSn())){
            parametrizedQuery.add(String.format("filter(fn: (r) => r[\"_measurement\"] == \"%s\")", influxDto.getDevSn()));
        }

        if (StringUtils.isNotEmpty(influxDto.getDevSn())){
            /*String devSnFilter = influxDto.getDevSn().stream().map(devSn -> String.format("r[\"devSn\"] == \"%s\"", devSn)).collect(Collectors.joining(" or "));

            System.out.println("拿到了:"+devSnFilter);*/
            parametrizedQuery.add(String.format("filter(fn: (r) => r[\"devSn\"] == \"%s\")", influxDto.getDevSn()));
            //parametrizedQuery.add(String.format("devSn(fn: (r) => %s)", devSnFilter));
        }

        if (influxDto.getFields()!=null&&influxDto.getFields().size()>0){
            String filedFilter = influxDto.getFields().stream().map(field -> String.format("r[\"_field\"] == \"%s\"", field)).collect(Collectors.joining(" or "));
            parametrizedQuery.add(String.format("filter(fn: (r) => %s)", filedFilter));
        }

        /**
         * 排序 按时间倒序
         */
        parametrizedQuery.add("sort(columns: [\"_time\"], desc: true)");

        /**
         * 分页
         */
        //parametrizedQuery.add(String.format("limit(n: %s, offset: %s)", pageSize, pageSize * (pageNum - 1)));

        Map<String, List<AttributeData>> result = new HashMap<>();
        parametrizedQuery.stream().forEach(System.out::println);
        List<FluxTable> tables = queryApi.query(String.join(" |> ", parametrizedQuery));
        DateTimeFormatter customFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        for (FluxTable table : tables) {

            List<List> list = table.getRecords().stream().map(record ->
                    Arrays.asList(record.getValue(), record.getTime().atZone(ZoneId.systemDefault()).toLocalDateTime())  // 日期直接转成本地的，前端拿去直接用
            ).collect(Collectors.toList());

            ArrayList<AttributeData> dataList = new ArrayList<>();
            list.forEach(s->{
                AttributeData data = new AttributeData();
                data.setValue(s.get(0).toString());
                data.setTheTime(((LocalDateTime)s.get(1)).format(customFormatter));
                dataList.add(data);
            });

            result.put(table.getRecords().get(0).getField(), dataList);
        }

        return result;

    }


    @Override
    public Page<AttributeData> historiesPage(Page<AttributeData> page, InfluxDto influxDto) {

        QueryApi queryApi = influxDBClient.getQueryApi();
        List<String> parametrizedQuery = new ArrayList<>();

        // 查询 Bucket
        parametrizedQuery.add(String.format("from(bucket: \"%s\")", influxDto.getBucket()));

        // 日期范围处理
        String startTime = influxDto.getStartTime();
        String endTime = influxDto.getEndTime();
        if (StringUtils.isNotEmpty(startTime) && StringUtils.isNotEmpty(endTime)) {
            parametrizedQuery.add(String.format("range(start: %s, stop: %s)",
                    DateUtils.subtract8Hours(startTime),
                    DateUtils.subtract8Hours(endTime)));
        } else {
            throw new IllegalArgumentException("StartTime and EndTime must not be null.");
        }

        // 测量名称过滤
        if (StringUtils.isNotEmpty(influxDto.getDevSn())) {
            parametrizedQuery.add(String.format("filter(fn: (r) => r[\"_measurement\"] == \"%s\")", influxDto.getDevSn()));
        }

        // 设备序列号过滤
        if (StringUtils.isNotEmpty(influxDto.getDevSn())) {
            parametrizedQuery.add(String.format("filter(fn: (r) => r[\"devSn\"] == \"%s\")", influxDto.getDevSn()));
        }

        // 字段过滤
        if (influxDto.getFields() != null && !influxDto.getFields().isEmpty()) {
            String fieldFilter = influxDto.getFields().stream()
                    .map(field -> String.format("r[\"_field\"] == \"%s\"", field))
                    .collect(Collectors.joining(" or "));
            parametrizedQuery.add(String.format("filter(fn: (r) => %s)", fieldFilter));
        }

        long totalRecords = getTotalRecords(parametrizedQuery, queryApi);



        // 排序
        parametrizedQuery.add("sort(columns: [\"_time\"], desc: true)");
        // 分页：使用 limit 和 offset
        long pageSize = page.getSize(); // 获取每页记录数
        long pageNum = page.getCurrent(); // 获取当前页码
        parametrizedQuery.add(String.format("limit(n: %d, offset: %d)", pageSize, pageSize * (pageNum - 1)));

        // 查询并处理结果
        List<FluxTable> tables = queryApi.query(String.join(" |> ", parametrizedQuery));
        List<AttributeData> dataList = new ArrayList<>();
        DateTimeFormatter customFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        for (FluxTable table : tables) {
            List<List> list = table.getRecords().stream().map(record ->
                    Arrays.asList(record.getValue(), record.getTime().atZone(ZoneId.systemDefault()).toLocalDateTime())
            ).collect(Collectors.toList());

            list.forEach(s -> {
                AttributeData data = new AttributeData();
                data.setValue(s.get(0).toString());
                data.setTheTime(((LocalDateTime)s.get(1)).format(customFormatter));
                dataList.add(data);
            });
        }

        // 计算总记录数
        //long totalRecords = dataList.size(); // 可根据实际需要调整，比如不再做 totalCounts 查询
        //int totalPages = (int) Math.ceil((double) totalRecords / pageSize);

        // 更新 Page 对象
        page.setRecords(dataList);
        page.setTotal(totalRecords);
        //page.setTotal(totalPages); // 假设你有这样的 setter 方法

        return page;
    }

    private long getTotalRecords(List<String> parametrizedQuery, QueryApi queryApi) {
        List<String> query = new ArrayList<>(parametrizedQuery);
        // 使用 count 函数计算总记录数
        query.add("aggregateWindow(every: 1s, fn: count, createEmpty: false)"); // 可以调整聚合窗口的大小，或使用 count() 函数直观的查询
        List<FluxTable> tables = queryApi.query(String.join(" |> ", query));
        // 获取总记录数
        long totalRecords = 0;
        if (!tables.isEmpty()) {
            for (FluxTable table : tables) {
                totalRecords += table.getRecords().size(); // 获取 count 结果
            }
        }

        return totalRecords;
    }


    @Override
    public AttributeData getNew(InfluxNewDto influxNewDto){
        QueryApi queryApi = influxDBClient.getQueryApi();

        List<String> parametrizedQuery = new ArrayList<>();

        Calendar calendar = Calendar.getInstance();

        calendar.add(Calendar.MONTH, -1);

        parametrizedQuery.add(String.format("from(bucket: \"%s\")", influxNewDto.getBucket()));

        try {
            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:SS");
            Date start = format.parse("2024-07-01 00:00:00");
            Date end = format.parse("2024-07-30 00:00:00");


            parametrizedQuery.add(String.format("range(start: %s, stop: %s)",DateUtils.subtract8Hours(start) ,DateUtils.subtract8Hours(end)));
        }catch (Exception e){

        }

        // toInfluxDate 就是给小时减了8小时，其实在前端处理更合理，只是我们系统没有国际化的需求，算是给前端偷懒了

        if (StringUtils.isNotEmpty(influxNewDto.getDevSn())) {
            parametrizedQuery.add(String.format("filter(fn: (r) => r[\"_measurement\"] == \"%s\")", influxNewDto.getDevSn()));

        }

        if (StringUtils.isNotEmpty(influxNewDto.getDevSn())){
            parametrizedQuery.add(String.format("filter(fn: (r) => r[\"devSn\"] == \"%s\")", influxNewDto.getDevSn()));

        }

        if (StringUtils.isNotEmpty(influxNewDto.getField())){
            parametrizedQuery.add(String.format("filter(fn: (r) => r[\"_field\"] == \"%s\")", influxNewDto.getField()));

        }

        parametrizedQuery.add("sort(columns: [\"_time\"], desc: true)");

        parametrizedQuery.add(String.format("limit(n: 1, offset: 0)"));

        Map<String, List<List>> result = new HashMap<>();

        List<FluxTable> tables = queryApi.query(String.join(" |> ", parametrizedQuery));

        for (FluxTable table : tables) {

            List<List> list = table.getRecords().stream().map(record ->
                    Arrays.asList(record.getValue(), record.getTime().atZone(ZoneId.systemDefault()).toLocalDateTime())  // 日期直接转成本地的，前端拿去直接用
            ).collect(Collectors.toList());

            result.put(table.getRecords().get(0).getField(), list);
        }

        if (result.size()>0){
            List list = result.get(influxNewDto.getField()).get(0);
            AttributeData attributeData = new AttributeData();
            attributeData.setValue(list.get(0).toString());
            attributeData.setTheTime(DateUtils.getDateByInflux(list.get(1).toString()));
            return  attributeData;
        }
        return null;
    }

    private static Map<String, Object> extracted(JSONObject items) {
        return items.keySet().stream()
                .collect(Collectors.toMap(Function.identity(), key -> {
                    JSONObject boj = items.getJSONObject(key);
                    return boj.get("value");
                }));
    }
}
